package cn.tedu.csmall.cart.webapi.kafka;

import cn.tedu.csmall.commons.pojo.cart.model.Cart;
import com.google.gson.Gson;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

// 当前类是来接收Kafka发送的消息用的
// 要求将这个类也保存到Spring容器中,因为SpringTemplate使用spring容器中的对象
@Component
public class Consumer {

    // spring-Kafka接收消息,使用了"监听机制"
    // 框架设计了一条线程,实时关注Kafka话题接收消息的情况
    // 我们指定一个话题名称(myCart),这个话题一旦接收消息,监听线程就会通知当前方法
    @KafkaListener(topics = Producer.TOPIC_KEY)
    // 上面就是设置监听器的注解,指定了myCart这个话题名称
    // 当kafka中出现myCart话题的消息时,监听器会自动调用下面的方法
    // 方法的参数和返回值是指定的不能修改
    public void received(ConsumerRecord<String,String> record){
        // 返回值必须是void 参数必须是ConsumerRecord类型
        // 泛型指定对应发送时的泛型<[话题类型],[消息类型]>
        // 这个参数record就是消息的发送者发送到kafka的内容,由监听器自动赋值
        // 从消息对象中获得消息内容
        String json=record.value();
        // json就是发送来的消息内容,它可能是这样值:{"id":"1","userId":"UU100","price":"50",....}
        // 下面再使用Gson工具类将json格式字符串转换为java对象
        Gson gson=new Gson();
        Cart cart=gson.fromJson(json,Cart.class);
        // 转换完成,输出Cart对象
        System.out.println(cart);
    }





}
